package com.nba;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class ChampionCountToFile {

    public static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
                throws IOException, InterruptedException {
            String[] split = value.toString().split(",");
            //取出冠军队的名称
            String champion = split[2];
            String area = "";//获取分区标志
            if (split.length == 7) { //带有分区标志的
                area = split[6]; //得到分区标志
            }
            Text newKey = new Text();
            if (area != null) {
                newKey.set(champion + "," + area); //有分区标志的，key如纽约尼克斯队,E
            } else {
                newKey.set(champion);//无分区标志
            }

            context.write(newKey, new IntWritable(1));  //如：写入的就是：如纽约尼克斯队,E    1

        }
    }

    static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        /**
         * 针对冠军队的数量进行统计
         *
         * @param key
         * @param values
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable value : values) {
                sum += value.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }

    static class MyOutputFormat extends FileOutputFormat<Text, IntWritable> {
        @Override
        public RecordWriter<Text, IntWritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
            return new MyRecordWriter();
        }
    }

    static class MyRecordWriter extends RecordWriter<Text, IntWritable> {
        FSDataOutputStream areaEast;
        FSDataOutputStream areaWest;
        FSDataOutputStream noArea;

        public MyRecordWriter() throws IOException {
            Configuration conf = new Configuration();
            FileSystem fs = FileSystem.get(conf);
            areaEast = fs.create(new Path("file/NBA/task3/areaeast.txt"));
            areaWest = fs.create(new Path("file/NBA/task3/areawast.txt"));
            noArea = fs.create(new Path("file/NBA/task3/noarea.txt"));
        }

        public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException, ArrayIndexOutOfBoundsException {
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);

            job.setJarByClass(ChampionCountToFile.class);
            job.setJobName("task3");

            job.setMapperClass(MyMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);

            job.setCombinerClass(MyReducer.class);
            job.setOutputFormatClass(MyOutputFormat.class);
            job.setReducerClass(MyReducer.class);
            job.setOutputKeyClass(IntWritable.class);
            job.setOutputValueClass(IntWritable.class);
            job.setOutputKeyClass(Text.class);
            FileInputFormat.setInputPaths(job, new Path("file/NBA/dataclear/out/part-m-00000"));

            FileOutputFormat.setOutputPath(job, new Path("file/NBA/task3"));
            //FileInputFormat.setInputPaths(job, new Path(args[0]));
            //FileOutputFormat.setOutputPath(job, new Path(args[1]));
            System.out.println(job.waitForCompletion(true));

        }

        @Override
        public void write(Text key, IntWritable value) throws IOException, InterruptedException {
            String[] split = key.toString().split(",");
            if (split.length > 1) {
                String area = split[1];
                if (area.equals("E")) {
                    areaEast.write((key + "    " + value.get()).getBytes());
                    areaEast.write("\n".getBytes());
                } else {
                    areaWest.write((key + "    " + value.get()).getBytes());
                    areaWest.write("\n".getBytes());
                }
            } else {
                noArea.write((key + "    " + value.get()).getBytes());
                noArea.write("\n".getBytes());
            }
        }

        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            IOUtils.closeStream(areaEast);
            IOUtils.closeStream(areaWest);
            IOUtils.closeStream(noArea);
        }
    }
}